package com.xiao.dynamic_rule.functions;

import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.ResultBean;
import com.xiao.dynamic_rule.pojo.RuleParam;
import com.xiao.dynamic_rule.service.*;
import com.xiao.dynamic_rule.utils.RuleSimulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 规则处理主逻辑
 */
public class RuleProcessFunction extends KeyedProcessFunction<String, LogBean, ResultBean> {
    // 创建解析规则的类
    private UserProfileQueryService userProfileQueryService;
    private UserActionCountQueryService userActionCountQueryServiceState;
    private UserActionSequenceQueryService userActionSequenceQueryServiceState;
    private RuleParam ruleParam;
    private ListState<LogBean> listState;

    /**
     * 创建解析规则实现类
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        userProfileQueryService = new UserProfileQueryServiceHbaseImpl();

        userActionCountQueryServiceState = new UserActionCountQueryServiceStateImpl();
        userActionSequenceQueryServiceState = new UserActionSequenceQueryServiceStateImpl();

        userActionCountQueryServiceState = new UserActionCountQueryServiceClickHouseImpl();
        userActionSequenceQueryServiceState = new UserActionSequenceQueryServiceClickhouseImpl();
        ruleParam = RuleSimulator.getRuleParam();
        //将数据存入state中
        listState = getRuntimeContext().getListState(new ListStateDescriptor<LogBean>("eventState", LogBean.class));

    }

    @Override
    public void processElement(LogBean logBean, Context context, Collector<ResultBean> out) throws Exception {

        //判断是否满足触发条件
        if (ruleParam.getTriggerParam().getEventId().equals(logBean.getEventId())) {
            //画像条件
            boolean profileMatch = userProfileQueryService.judgeQueryProfileCondition(logBean.getDeviceId(), ruleParam);
            if (!profileMatch) return;

            //行为次数条件
            boolean countMatch = userActionCountQueryServiceState.queryActionCounts("",listState, ruleParam);
            if (!countMatch) return;

            //行为序列条件
            boolean sequenceMatch = userActionSequenceQueryServiceState.queryActionSequence("",listState, ruleParam);
            if (!sequenceMatch) return;
        }

        // 说明都满足，返回成功结果集
        out.collect(new ResultBean(ruleParam.getRuleId(), logBean.getDeviceId(), logBean.getTimeStamp()));
    }
}
